package cn.jly.flink.custom;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Properties;

/**
 * sink to mysql
 *
 * @author lanyangji
 */
public class SinkToMysqlApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties props = new Properties();
        props.put("bootstrap.servers", "hadoop101:9092");
        props.put("zookeeper.connect", "hadoop101:2181");
        props.put("group.id", "student-group");
        //key 反序列化
        props.put("key.deserializer", StringDeserializer.class.getName());
        //value 反序列化
        props.put("value.deserializer", StringDeserializer.class.getName());
        props.put("auto.offset.reset", "latest");

        FlinkKafkaConsumer011<String> consumer = new FlinkKafkaConsumer011<>("student", new SimpleStringSchema(), props);
        consumer.setStartFromEarliest();

        SingleOutputStreamOperator<Student> studentDataStream =
                env.addSource(consumer)
                        .setParallelism(1)
                        .map(str -> JSON.parseObject(str, Student.class));

        // sink to mysql
        studentDataStream.addSink(new SinkToMysql());

        env.execute("SinkToMysql");
    }
}
